package threadpool;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.RecursiveTask;
import java.util.stream.Collectors;
import java.util.stream.Stream;

/**
 * 它的模型大致是这样的：
 * 线程池中的每个线程都有自己的工作队列
 * （PS：这一点和ThreadPoolExecutor不同，ThreadPoolExecutor是所有线程公用一个工作队列，所有线程都从这个工作队列中取任务）
 * ，当自己队列中的任务都完成以后，会从其它线程的工作队列中偷一个任务执行，这样可以充分利用资源。
 * <p>
 * 每个线程都有自己维护的任务队列
 * <p>
 * <p>
 * 由于线程池的优化，提交的任务和线程数量并不是一对一的关系。
 * 在绝大多数情况下，一个物理线程实际上是需要处理多个逻辑任务的。
 * 因此，每个线程必然需要拥有一个任务队列。因此，在实际执行过程中，
 * 可能遇到这么一种情况：线程A已经把自己的任务都处理完了，
 * 而线程B还有一堆任务等着处理，
 * 此时，线程A就会“帮助” 线程B，从线程 B的任务队列中拿一个任务来处理，
 * 尽可能的达到平衡。值得注意的是：当线程试图帮助别人时，总是从任务队列的底部开始拿数据，
 * 而线程试图执行自己的任务时，则从相反的顶部开始拿。因此这种行为也十分有利于避免数据竞争。
 * 这种行为也叫做： 工作窃取
 * <p>
 * <p>
 * <p>
 * <p>
 * 你可以向 ForkJoinPool 线程池提交一个 ForkJoinTask 任务。
 * 所谓 ForkJoinTask 任务就是支持 fork （） 分解以及 join（）等待的任务。
 * ForkJoinTask 有两个重要的子类，RecursiveAction 和 RecursiveTask。
 * 他们分别表示没有返回值的任务和可以携带返回值的任务。有点像 Runnable 和 Callable。
 * <p>
 * <p>
 * Fork/Join 核心思想：分而治之
 * 著名的 MapReduce 也是这个思想。将任务进行分解，然后合并所有的结果。
 *
 *
 *
 *
 *
 * 主要方法：
 *
 * fork()    在当前线程运行的线程池中安排一个异步执行。简单的理解就是再创建一个子任务。
 * join()    当任务完成的时候返回计算结果。
 * invoke()    开始执行任务，如果必要，等待计算完成。
 *
 *
 *
 * JDK8 STREAM 并行流 的底层实现也是ForkJoin
 *
 */
public class T07_ForkJoinPoolTest {
    static List<Integer> pages = Stream.iterate(1, item -> item + 1).limit(1000000).collect(Collectors.toList());
    List<Map<String, String>> dataInfoList = new ArrayList<>();

    public static void main(String[] args) throws ExecutionException, InterruptedException {

        ForkJoinPool forkJoinPool = new ForkJoinPool();
        SpiderTask spiderTask = new SpiderTask(0, pages.size());
        forkJoinPool.execute(spiderTask);
        Integer join = spiderTask.join();
//        Integer join = spiderTask.get();
//        Integer join = spiderTask.invoke();
        System.out.println(join);
        System.out.println(pages.stream().reduce(Integer::sum).get().intValue());

//        JDK8 STREAM 并行流 的底层实现也是ForkJoin
//        pages.parallelStream()


    }


    static class SpiderTask extends RecursiveTask<Integer> {
        int start;
        int mid;
        int end;
        int GROUP_SIZE = 50000;

        public SpiderTask(int start, int end) {
            this.start = start;
            this.end = end;
            this.mid = start + (end - start) / 2;
        }

        @Override
        protected Integer compute() {
            if (end - start <= GROUP_SIZE) {
                int num = 0;
                for (int i = start; i < end; i++) num += pages.get(i);
                System.out.println("start:" + start + "--" + end + "--" + num);
                return num;

            } else {

                SpiderTask subTask = new SpiderTask(start, mid);
                SpiderTask subTask2 = new SpiderTask(mid, end);
                subTask.fork();
                subTask2.fork();
                return subTask.join() + subTask2.join();

            }


        }
    }

}
